package com.run.wz.center.hbase.actionbuilder.impl;

import java.io.IOException;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Calendar;
import java.util.Collections;
import java.util.Comparator;
import java.util.HashSet;
import java.util.List;
import java.util.Map.Entry;
import java.util.NavigableMap;
import java.util.NavigableSet;
import java.util.Set;
import java.util.TreeMap;
import java.util.TreeSet;

import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.regionserver.HRegion;

import com.run.wz.center.hbase.HBaseBatchUtil;
import com.run.wz.center.util.ObjectType;
import com.run.wz.center.utils.GroupType;
import com.run.wz.center.utils.ProtocolType;
/**
 * CF是"天"的ScanBuilder,这里认为这样的表可以保留两种方法的实现
 * 1.根据开始结束时间去构建不同的CF并根据配置好的策略将不同的CF分配到不同的
 * 表的不同的scan中.
 * 2.由于CF是"天" 所以认为表名会包含时间后缀.
 * 
 * Map的Key的表名、Value中的TreeSet<Scan>、Scan中的CF都是按照倒序
 * 排序的。
 * 针对一次查询请求构建scan，根据时间和协议构建结果可能是多个表。
 * @author chenxu 2012-9-6 下午1:13:25
 *
 */
public class CFByDayScanBuilder extends ScanBuilder  {
	
	private long startTime;
	private long endTime;
	private int groupSize;
	protected TreeSet<ProtocolType> protocolTypeSet;
	public CFByDayScanBuilder(long startTime, long endTime,
			ObjectType objectType, String oid,TreeSet<ProtocolType> protocolTypeSet) {
		super(oid, objectType);
		this.startTime = startTime;
		this.endTime = endTime;
		this.protocolTypeSet = protocolTypeSet==null || protocolTypeSet.isEmpty() ? HBaseBatchUtil.PROTOCOL_TYPESET : protocolTypeSet;
		this.oid = oid;
		this.objectType = objectType;
		groupSize = 10;
	}
	/**
	 * 根据构造入参，构建每个表的scan集合（根据时间和协议构建结果可能是多个表）
	 * 
	 * 构建逻辑遵循以下步骤：
	 * 1、根据入参（开始、结束时间，协议集合，对象类型）先计算出来可能需要查询哪些表。
	 * 2、根据构造出来的表名取出事先缓存这个表的 {@link HTableDescriptor}
	 * 根据所有入参去构建scan（会涉及到——将CF、协议分配到不同的scan中），最终返回
	 * 一个scan集合。
	 * 
	 * @return 每个表的scan集合
	 * @see HTableDescriptor
	 * @see HBaseBatchUtil
	 */
	public NavigableMap<String, TreeSet<Scan>> build() {
		
		TreeMap<String, TreeSet<Scan>> retMap = new TreeMap<String, TreeSet<Scan>>();
		for(Entry<String,TreeSet<ProtocolType>> entry : buildTableNames().entrySet()) {
			HTableDescriptor nextStepHTD = HBaseBatchUtil.getHTD(entry.getKey());
			try {
				if(null!=nextStepHTD) {
					retMap.put(entry.getKey(), buildScans(nextStepHTD, entry.getValue()));
				} else {
					logger.warn(this+"tableName["+entry.getKey()+"] no HTD so skip it");
				}
			} catch (ParseException shouldnote) {
				logger.warn("shouldnot occur exception, check carfully!!!", shouldnote); continue;
			}
		}
		return retMap.descendingMap();
	}
	
	/**
	 * 1、构建scan集合，构建出来的scan是符合入参 htd 的实际情况的（主要是CF是否存在）
	 * 2、在构建的时候可能会将多个CF分配的一个scan中，由于CF是按照“天”分的，hbase默认
	 * 是按照升序去逐个检索CF的，为了保证业务在查询的时候可以获得最新的数据手动的将scan的
	 * family map 做了倒序排序处理； 同时返回的集合也做了倒序处理。
	 * 
	 * chenxu 2012-9-12 上午10:56:08
	 * @param htd
	 * @return 倒序排序的的scan集合，可能为空（empty），但是不会为NULL
	 * @throws ParseException 处理表名的时候可能会抛出这个异常
	 * @see HRegion.RegionScannerImpl
	 */
	public TreeSet<Scan> buildScans(HTableDescriptor htd,TreeSet<ProtocolType> protocolTypeSet) throws ParseException {
		
		TreeSet<Scan> scans = new TreeSet<Scan>(new Comparator<Scan>() {

			@Override
			public int compare(Scan o1, Scan o2) {
				return ((byte[])((TreeMap<byte [], NavigableSet<byte []>>)o1.getFamilyMap()).firstKey())[0]
						-
					   ((byte[])((TreeMap<byte [], NavigableSet<byte []>>)o2.getFamilyMap()).firstKey())[0];
			}
		}) ;
		
		String tableName = htd.getNameAsString();
		Set<byte[]> familiesKeys = htd.getFamiliesKeys();
		String[] split = tableName.split(HBaseBatchUtil.DEFAULT_CONTACT);
		String dateStr = split[split.length-1];
		
		long minTR = 0;
		long maxTR = 0;
		long tableMinTR = new SimpleDateFormat(HBaseBatchUtil.DEFAULT_DATEFORMAT).parse(dateStr).getTime();
		Calendar instance = Calendar.getInstance();
		instance.setTimeInMillis(tableMinTR);
		//这个表还有多少个family就往上增加几天-1，因为已经有一天的基数了
		instance.add(Calendar.DAY_OF_MONTH, familiesKeys.size()-1);

		instance.set(Calendar.HOUR_OF_DAY, 23);
		instance.set(Calendar.MINUTE, 59);
		instance.set(Calendar.SECOND, 59);
		long tableMaxTR = instance.getTimeInMillis();
		
		minTR = Math.max(tableMinTR, startTime);//表的最小时间和 starttime两个谁大谁就是这张表的第一个scan的 minTimeRange
		maxTR = Math.min(tableMaxTR, endTime);//表的最大时间和endTime两个谁最小谁就是这张表的最后一个scan的maxTimeTange
			
		
		Calendar maxC = Calendar.getInstance();
		maxC.setTimeInMillis(maxTR);
		int maxD = maxC.get(Calendar.DAY_OF_MONTH);
		Calendar minC = Calendar.getInstance();
		minC.setTimeInMillis(minTR);
		int minD = minC.get(Calendar.DAY_OF_MONTH);
		int days = maxD-minD+1;
				
		long tempMinTR = minTR;
		long tempMaxTR = minTR;
		int i=0; 
		int batch = protocolTypeSet.size();
		Scan scan  = buildScan();
		do{
			//1、天数大于groupSize的话，每满一组new一个scan，或者已经到了整数组后的余数。2、如果天数小于groupSize的话到了循环完天数再new一个scan就可以
			if(
				(groupSize<=days && ((i%groupSize==0 && i!=0) 
				|| (((days%groupSize)==(i%groupSize))&&((i/groupSize)==(days/groupSize))))) 
				|| (groupSize>days && i==days)) {
				scan.setBatch(batch);
				scans.add(scan);
				try {
					scan.setTimeRange(tempMinTR, tempMaxTR);
				} catch (IOException nevere) {
					
				}
				scan = buildScan();
				tempMinTR = minC.getTimeInMillis();
			}
			byte[] family = new byte[]{(byte)(HBaseBatchUtil.COLUMN_NAME[minC.get(Calendar.DAY_OF_MONTH)])};
			if(familiesKeys.contains(family))
				for(ProtocolType pt : protocolTypeSet) {
					scan.addColumn(family,pt.getBytes());
				}
			else 
				logger.warn(this+"table["+tableName+"]'s CF["+family[0]+"] not exist so skip it ");
			minC.set(Calendar.HOUR_OF_DAY, 23);
			minC.set(Calendar.MINUTE, 59);
			minC.set(Calendar.SECOND, 59);
			
			/**
			 * 如果是最后一天，且最后一天的查询的入参的最大时间不到23:59:59秒，则tempMaxTR需要设为指定的时间
			 * 86399000 = 23:59:59 - 00:00:00
			 */
			if(maxTR - minC.getTimeInMillis() > 86399000) {
				tempMaxTR = minC.getTimeInMillis()>maxTR? minC.getTimeInMillis() : Math.min(minC.getTimeInMillis(), maxTR);//maxTR可能还没有到一天的最后一秒
			} else {
				tempMaxTR = maxTR;
			}
			minC.add(Calendar.SECOND, 1);//下一天
		} while(i++<days && tempMaxTR<=maxTR);
		//将scan的CF倒序
		for(Scan scanTemp : scans) {
			scanTemp.setFamilyMap(((TreeMap<byte[],NavigableSet<byte[]>>)scanTemp.getFamilyMap()).descendingMap());
		}
		
		return (TreeSet<Scan>)scans.descendingSet();
	}
	
	
	
	/**
	 * 根据指定的时间范围时间格式和连接符构建日期后缀，支持最小粒度为天
	 * @return 日期后缀字符串集合
	 * 
	 * modify by chenxu @ 2012-09-12
	 */
	public List<String> buildDateSuffix() {
		Set<String> suffixsTemp = new HashSet<String>();
		List<String> suffixs = new ArrayList<String>();

		long[] timestamps = new long[]{startTime, endTime};
		//时间区间对应多个表
		long startTime = Math.min(timestamps[0], timestamps[1]);
		long endTime = Math.max(timestamps[0], timestamps[1]);
		SimpleDateFormat sdf = new SimpleDateFormat(HBaseBatchUtil.DEFAULT_DATEFORMAT);
		Calendar instance = Calendar.getInstance();
		instance.setTimeInMillis(startTime);
		do{
			String suffix = HBaseBatchUtil.DEFAULT_CONTACT+sdf.format(instance.getTime());
			if(suffixsTemp.add(suffix))
				suffixs.add(suffix);
			instance.add(Calendar.DAY_OF_MONTH, 1);//之所以以天的粒度增加时考虑可能会出现“天”类型的后缀
		}while(instance.getTimeInMillis()<endTime);
	
		try {
			Collections.sort(suffixs);
		} finally {
			suffixsTemp.clear();
			suffixsTemp = null;
		}
		
		return suffixs;
	}
	
	public TreeMap<String, TreeSet<ProtocolType>> buildTableNames() {
		TreeMap<String,TreeSet<ProtocolType>> tableNames = new TreeMap<String,TreeSet<ProtocolType>>();
	
		if(objectType.needGroup()) {
			TreeMap<String, TreeSet<ProtocolType>> groupPTSetByGroup = GroupType.groupPTSetByGroup(protocolTypeSet);
			for(Entry<String, TreeSet<ProtocolType>> entry : groupPTSetByGroup.entrySet()) {
				for(String dateSuffix : buildDateSuffix()) {
					tableNames.put(objectType.getIndexTableNamePrefix()+entry.getKey()+dateSuffix, entry.getValue());
				}
			}
		} else {
			for(String dateSuffix : buildDateSuffix()) {
				tableNames.put(objectType.getIndexTableNamePrefix()+dateSuffix, protocolTypeSet);
			}
		}
		return tableNames;
	}
	@Override
	public String toString() {
		return "CFByDayScanBuilder [startTime=" + startTime + ", endTime="
				+ endTime + ", groupSize=" + groupSize + ", oid=" + oid + ", objectType="
				+ objectType + ", protocolTypeSet="
						+ protocolTypeSet + "]";
	}

}